﻿using System;
using System.Collections.Generic;
using System.Linq;
using CoreLibrary.Collection;
using CoreLibrary.Concurrent;
using CoreLibrary.Scheduler;
using CoreLibrary.Service;
using NLog;

namespace CoreLibrary.Messaging {

    public class MemoryQueue : ServiceBase, IMessageQueue {

        public ISchedulerService SchedulerService { get; set; }
        public ExceptionHandler  UncaughtExceptionHandler { get; set; }
        public int MaxThreadCount { get; set; }
        
        private AsyncBlockingTaskQueue _taskTaskQueue;
        private MessageQueueStatus _queueStatus = MessageQueueStatus.Stopped;

        private readonly CopyOnWriteList<IMessageHandler> _messageHandlers = new CopyOnWriteList<IMessageHandler>();
        private readonly ReadWriteDictionary<int, IMessageHandler> _msgHandlerMap = new ReadWriteDictionary<int,IMessageHandler>();  
        
        public MemoryQueue() {
            MaxThreadCount = 100;
        }

        protected override void OnInit() {
            Asserts.True(MaxThreadCount > 0,  "Max thread count must be greater than 0");
            Asserts.NotNull(SchedulerService, "Scheduler service is not in place");
            _taskTaskQueue = new AsyncBlockingTaskQueue(MaxThreadCount) {
                Executor = HandleMessage,
                ErrorHandler = HandleError
            };
        }

        private void HandleMessage(object task, AsyncBlockingTaskQueue.CompletionCallback completionCallback) {
            var message = (Message)task;
            try {
                var updated = UpdateMessageToProcessing(message);
                if (!updated) {
                    return;
                }

                MessageHandlingCallback callback = null;
                foreach (var messageHandler in _messageHandlers) {
                    if (messageHandler.CanHandleMessage(message)) {
                        callback = new MessageHandlingCallback(this, message);
                        messageHandler.HandleMessage(message, callback);
                        break;
                    }
                }

                if (callback == null) {
                    if (_queueStatus != MessageQueueStatus.Stopping &&
                        _queueStatus != MessageQueueStatus.Stopped) {
                        UpdateMessageToPending(message);
                    }
                }
            } 
            catch (Exception error) {
                HandleError(message, error);
                if (UncaughtExceptionHandler != null) {
                    try {
                        UncaughtExceptionHandler(error);
                    } catch (Exception failure) {
                        LogManager.GetCurrentClassLogger().ErrorException("Message queue uncaught exception handling failed: " + failure.Message, failure);
                    }
                } else {
                    LogManager.GetCurrentClassLogger().ErrorException("Processing message failed: " + error.Message, error);
                }
            } 
            finally {
                completionCallback();
            }
        }

        private void HandleError(object task, Exception error) {
            if (_queueStatus == MessageQueueStatus.Stopping ||
                _queueStatus == MessageQueueStatus.Stopped) {
                return;
            }
        }

        protected override void OnDestroy() {}
        protected override void OnStart() {
            lock (this) {
                if (_queueStatus == MessageQueueStatus.Started) {
                    return;
                }
                _taskTaskQueue.Start();
                _queueStatus = MessageQueueStatus.Started;
            }
        }

        protected override void OnStop() {
            lock (this) {
                if (_queueStatus == MessageQueueStatus.Stopping ||
                    _queueStatus == MessageQueueStatus.Stopped) {
                    return;
                }
                _taskTaskQueue.Stop();
                _queueStatus = MessageQueueStatus.Stopping;
            }
        }

        public void InsertMessageHandler(IMessageHandler handler) {
            if (!_messageHandlers.Contains(handler)) {
                 _messageHandlers.Add(handler);
            }
        }

        public void RemoveMessageHandler(IMessageHandler handler) {
            _messageHandlers.Remove(handler);
        }

        public SendMessageBatchResponse SendMessageBatch(SendMessageBatchRequest request) {
            throw new NotImplementedException();
        }

        public List<IMessageHandler> ListMessageHandlers() {
            return _messageHandlers.GetCopyForRead();
        }

        public int GetMessageCount() {
            return _taskTaskQueue.Count();
        }

        public int GetMaxActiveMessageCount() {
            return _taskTaskQueue.MaxThread;
        }

        public int GetProcessingMessageCount() {
            return _taskTaskQueue.CountActive();
        }

        public int GetAwaitingMessageCount() {
            return _taskTaskQueue.CountInactive();
        }

        public MessageQueueStatus GetQueueState() {
            lock (this) {
                if (_queueStatus == MessageQueueStatus.Stopping && _taskTaskQueue.CountActive() == 0) {
                    _queueStatus = MessageQueueStatus.Stopped;
                }
                return _queueStatus;
            }
        }

        public List<Message> ListMessages(int offset, int max) {
            var tasks = _taskTaskQueue.Get(offset, max);
            return tasks.Cast<Message>().ToList();
        }

        public Message RemoveMessage(string messageId) {
            _taskTaskQueue.Remove(messageId);
            return null;
        }

        public Message RetryMessage(string messageId, string errorMsg) {
            throw new System.NotImplementedException();
        }

        private void EnqueueMessage(object task) {
            var message = (Message)task;
            _taskTaskQueue.Enqueue(message, GetTaskPriority(message.Priority));
        }

        public SendMessageResponse SendMessage(SendMessageRequest request) {
            lock (this) {
                var message = new Message {
                    Id            = Guid.NewGuid().ToString(),
                    ApplicationId = request.ApplicationId,
                    Data          = request.Body,
                    Status        = MessageStatus.Enqueued,
                    Fault         = "",
                    DateCreated   = request.DateCreated,
                    DateScheduled = request.DateScheduled,
                    Priority      = request.Priority
                };

                var now = DateTime.Now;
                try {
                    var timeSpan = message.DateScheduled.Subtract(now);
                    if (timeSpan.TotalMinutes >= 3) {
                        SchedulerService.Schedule(EnqueueMessage, message, timeSpan);
                    }
                    else {
                        _taskTaskQueue.Enqueue(message, GetTaskPriority(message.Priority));
                    }
                }
                catch (Exception exception) {
                    LogManager.GetCurrentClassLogger().ErrorException("Scheduling message failed: " + exception.Message, exception);
                    return new SendMessageResponse {
                        Success      = false,
                        MessageId    = null,
                        ErrorMessage = "Scheduling message failed: " + exception.Message,
                    };
                }

                return new SendMessageResponse {
                    ErrorMessage = string.Empty,
                    MessageId    = message.Id,
                    Success      = true
                };
            }
        }

        private static AsyncBlockingTaskQueue.Priority GetTaskPriority(MessagePriority priority) {
            if (priority == MessagePriority.High) {
                return AsyncBlockingTaskQueue.Priority.High;
            }
            return AsyncBlockingTaskQueue.Priority.Normal;
        }

        private bool UpdateMessageToEnqueued(Message message) {
            message.Status = MessageStatus.Enqueued;
            return true;
        }

        private bool UpdateMessageToProcessing(Message message) {
            message.Status = MessageStatus.Processing;
            return true;
        }

        private bool UpdateMessageToFailed(Message message, string errorMsg) {
            message.Status = MessageStatus.Failed;
            message.Fault = StringUtil.SubString(errorMsg, 0, 1000);
            return true;
        }

        private bool UpdateMessageToFailed(Message message, Exception error) {
            return UpdateMessageToFailed(message, Util.GetString(error.Message));
        }

        private bool UpdateMessageToPending(Message message) {
            message.Status = MessageStatus.Pending;
            return true;
        }

        private bool UpdateMessageToProcessed(Message message) {
            message.Status = MessageStatus.Processed;
            return true;
        }

        private bool UpdateMessageToRetry(Message message, string errorMsg, DateTime dateScheduled) {
            message.Status = MessageStatus.Enqueued;
            message.Fault  = errorMsg;
            message.Retry += 1;
            message.DateScheduled = dateScheduled;
            _taskTaskQueue.Remove(message);
            return true;
        }

        private class MessageHandlingCallback : IMessageProcessingCallback {

            private readonly MemoryQueue _messageQueue;
            private readonly Message _message;

            public MessageHandlingCallback(MemoryQueue messageQueue, Message message) {
                _messageQueue = messageQueue;
                _message = message;
            }

            public void OnSuccess() {
                _messageQueue.UpdateMessageToProcessed(_message);
            }

            public void OnFailure(string errorMsg) {
                _messageQueue.UpdateMessageToFailed(_message, errorMsg);
            }

            public void OnRetry(string errorMsg, TimeSpan delay) {
                _messageQueue.UpdateMessageToRetry(_message, errorMsg, DateTime.Now.Add(delay));
                _messageQueue.SchedulerService.Schedule(_messageQueue.EnqueueMessage, _message, delay);
            }
        }
    }
}